/**
 * 
 */
package stc.skymobi.cassandra;

import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Date;

import me.prettyprint.cassandra.service.CassandraClient;
import me.prettyprint.cassandra.service.CassandraClientPool;
import me.prettyprint.cassandra.service.CassandraClientPoolFactory;
import me.prettyprint.cassandra.service.Keyspace;
import me.prettyprint.cassandra.service.PoolExhaustedException;
import me.prettyprint.cassandra.utils.StringUtils;

import org.apache.cassandra.thrift.SuperColumn;
import org.apache.cassandra.thrift.Column;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.commons.beanutils.PropertyUtils;
import org.apache.commons.collections.Closure;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author hp
 *
 */
public class CassandraAppender implements Closure {

    private static final Logger logger = LoggerFactory.getLogger(CassandraAppender.class);
    
	private	String			urlPort;
	private	String			keyspace;
	private	String			columnFamily;
	private	String			keyProperty;
	private	String[]			superColumnProperties = null;
    private Map<String, String> columnProperties = new HashMap<String, String>();
	
	private CassandraClientPool pool;
	private	Keyspace		ks;
	
	/* (non-Javadoc)
	 * @see org.apache.commons.collections.Closure#execute(java.lang.Object)
	 */
	public void execute(Object input) {
		
		try {
			String 	key = BeanUtils.getProperty(input, keyProperty);
			String	superColumnName = null;
			String 	prefix = "";
			
			for ( String property : this.superColumnProperties ) {
				superColumnName = prefix + BeanUtils.getProperty(input, property);
				prefix = superColumnName + "-";
			}
			
			List<Column> 	columnList 	= new ArrayList<Column>();
			
			for ( Map.Entry<String, String> entry : this.columnProperties.entrySet() ) {
				Object val = PropertyUtils.getProperty(input, entry.getKey() ) ;
				if ( null != val ) {
                    String value;
                    if(val instanceof Date){
                        value = Long.toString(((Date)val).getTime());
                    }
                    else{
                        value = val.toString();
                    }
					Column	column = new Column( 
							StringUtils.bytes(entry.getValue()),
							StringUtils.bytes(value),
							System.currentTimeMillis()
							);
					columnList.add(column);
				}
			}
			
			if ( null != superColumnName ) {
				SuperColumn	superColumn = new SuperColumn();
				superColumn.setName( StringUtils.bytes(superColumnName));
				superColumn.setColumns(columnList);
				
				Map<String, List<SuperColumn>> superColumnMap = 
					new HashMap<String, List<SuperColumn>>();
				List<SuperColumn> superColumnList = new ArrayList<SuperColumn>();
				superColumnList.add(superColumn);
				
				superColumnMap.put(this.columnFamily, superColumnList);
				ks.batchInsert(key, null, superColumnMap);
			}
			else {
				Map<String, List<Column>> columnMap = 
					new HashMap<String, List<Column>>();
				columnMap.put(this.columnFamily, columnList);
				ks.batchInsert(key, columnMap, null);
			}
		}
		catch (Exception e) {
			logger.error("execute:" + input, e);
		}
	}

	public void start() throws IllegalStateException, PoolExhaustedException, Exception {
		CassandraClientPool pool = CassandraClientPoolFactory.INSTANCE.get();
		CassandraClient client = pool.borrowClient(this.urlPort);
		ks = client.getKeyspace(this.keyspace);
	}
	
	public void stop() throws Exception {
		if ( null != pool && null != ks ) {
			pool.releaseKeyspace(ks);
		}
	}
	
	/**
	 * @return the urlPort
	 */
	public String getUrlPort() {
		return urlPort;
	}

	/**
	 * @param urlPort the urlPort to set
	 */
	public void setUrlPort(String urlPort) {
		this.urlPort = urlPort;
	}

	/**
	 * @return the keyspace
	 */
	public String getKeyspace() {
		return keyspace;
	}

	/**
	 * @param keyspace the keyspace to set
	 */
	public void setKeyspace(String keyspace) {
		this.keyspace = keyspace;
	}

	/**
	 * @return the keyProperty
	 */
	public String getKeyProperty() {
		return keyProperty;
	}

	/**
	 * @param keyProperty the keyProperty to set
	 */
	public void setKeyProperty(String keyProperty) {
		this.keyProperty = keyProperty;
	}

	/**
	 * @return the columnFamily
	 */
	public String getColumnFamily() {
		return columnFamily;
	}

	/**
	 * @param columnFamily the columnFamily to set
	 */
	public void setColumnFamily(String columnFamily) {
		this.columnFamily = columnFamily;
	}

	/**
	 * @return the superColumnProperties
	 */
	public List<String> getSuperColumnProperties() {
		return Arrays.asList( superColumnProperties );
	}

	/**
	 * @param superColumnProperties the superColumnProperties to set
	 */
	public void setSuperColumnProperties(List<String> superColumnProperties) {
		this.superColumnProperties = superColumnProperties.toArray(new String[0]);
	}

	/**
	 * @return the columnProperties
	 */
	public Map<String, String> getColumnProperties() {
		return Collections.unmodifiableMap(columnProperties);
	}

	/**
	 * @param columnProperties the columnProperties to set
	 */
	public void setColumnProperties(Map<String, String> columnProperties) {
        this.columnProperties.clear();
        
        for (Map.Entry<String, String> entry : columnProperties.entrySet()) {
            if ( null != entry.getValue() ) {
                this.columnProperties.put(entry.getKey(),  entry.getValue() );
            }
        }
	}
}
